package com.atguigu.flink.chapter06;

import com.atguigu.flink.bean.WaterSensor;
import com.atguigu.flink.chapter05.function.WaterSensorMapFunction;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;


/*
增量处理:来一个处理计算一个，时间一到输出结果
全量处理：来齐之后在处理计算，并输出结果
窗口处理函数：
    简单的
        sum  增量处理
        max min  增量处理
        maxBy minBy 增量处理
    复杂的
        reduce 增量处理

        aggrgate 增量处理 ----》可以做到输出和输入的类型不一致

        process   全量处理
 */
public class WindowProcessDemo1 {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 2000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);

        env.socketTextStream("hadoop102",8888)
                .map(new WaterSensorMapFunction())
                .keyBy(WaterSensor::toString)
                .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
                //.sum("vc")
                //.min("vc")
                //.max("vc")
                //.maxBy("vc")
                //.minBy("vc")
                /*.reduce(new ReduceFunction<WaterSensor>() {
                    @Override
                    public WaterSensor reduce(WaterSensor value1, WaterSensor value2) throws Exception {
                        System.out.println("xxxx");
                        value1.setVc(value1.getVc() + value2.getVc());
                        return value1;
                    }
                })
                 */
                /*
                参数1：输入
                参数2：中间结果
                参数3：输出
                 */
                .aggregate(new AggregateFunction<WaterSensor, Avg, Double>() {
                    //创建累加器（初始化：累加器）：当这个窗口的第一个元素到的时候执行一次
                    @Override
                    public Avg createAccumulator() {
                        System.out.println("WindowProcessDemo1.createAccumulator");
                        return new Avg();
                    }

                    //累加：每来一条元素，触发一次
                    @Override
                    public Avg add(WaterSensor value, Avg acc) {
                        System.out.println("WindowProcessDemo1.add");
                        acc.vcSum += value.getVc();
                        acc.count++;
                        return null;
                    }

                    //返回最终结果
                    @Override
                    public Double getResult(Avg acc) {
                        System.out.println("WindowProcessDemo1.getResult");
                        return acc.vcSum * 1.0 / acc.count;
                    }

                    //累加器的合并：这个方法只有当窗口是会话窗口的时候才会触发
                    //如果不是会话窗口，就不用实现
                    @Override
                    public Avg merge(Avg avg, Avg acc1) {
                        System.out.println("WindowProcessDemo1.merge");
                        return null;
                    }
                })
                .print();

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
class Avg {
    public Integer vcSum = 0;
    public Integer count = 0;
}